#include "DatagramAcceptChannel.h"
#include <unistd.h>
#include "EventLoop.h"
#include "IpV4Address.h"
#include "IpV6Address.h"
#include "Manager.h"
#include "MessageHeader.h"
#include "PlatLogger.h"
#include "SocketOpt.h"
#include "UdpSocket.h"
#include "Util.h"
#include "XnccType.h"

namespace xncc {
namespace foundation {

    DatagramAcceptChannel::DatagramAcceptChannel(EventLoop*                   loop,
                                                 std::shared_ptr<UdpSocket>   serverSocket,
                                                 const std::set<std::string>& blacklist)
        : Channel(loop, serverSocket->fd()),
          HEADER_LEN(static_cast<int32_t>(sizeof(msg::Header))),
          serverSocket_(std::move(serverSocket)),
          blacklist_(blacklist)
    {
        TRACE_INFO << "create DatagramAcceptChannel bind loop id:" << loop->id() << " fd:" << serverSocket_->fd()
                   << " loop type:" << types::IO_MULTI_MODE::toString(loop->mode());
    }

    DatagramAcceptChannel::~DatagramAcceptChannel() { ::close(fd()); }

    void DatagramAcceptChannel::handleWrite() {}
    /// IO线程，收到消息之后，就是构建消息，然后投递到订阅消息的线程
    void DatagramAcceptChannel::handleRead()
    {
        char                buf[65536]   = {0};
        struct sockaddr_in6 peerAddr     = {};
        auto*               peerSockaddr = reinterpret_cast<struct sockaddr*>(&peerAddr);
        unsigned int        sockLen      = sizeof(struct sockaddr);
        ssize_t n = ::recvfrom(fd(), buf, sizeof(buf), 0, peerSockaddr, reinterpret_cast<socklen_t*>(&sockLen));

        int clientSockfd = SocketOpt::createNonblockingOrDie(peerSockaddr->sa_family, IPPROTO_UDP, SOCK_DGRAM);
        SocketOpt::setReuseAddr(true, clientSockfd);
        SocketOpt::setReusePort(true, clientSockfd);
        serverSocket_->bind(clientSockfd);  // 绑定本地地址
        auto ret = ::connect(clientSockfd, peerSockaddr, sockLen);
        (void)ret;

        std::shared_ptr<XnccAddress> clientAddr;
        const auto                   connId = shared::CommonBusi::generateConnId();
        if (peerSockaddr->sa_family == AF_INET) {
            auto v4PAdr = std::make_shared<IpV4Address>(IPPROTO_UDP);
            v4PAdr->setSockAddr(*SocketOpt::sockaddr_in_cast(peerSockaddr));
            clientAddr = v4PAdr;
        }
        else if (peerSockaddr->sa_family == AF_INET6) {
            auto v6PAdr = std::make_shared<IpV6Address>(IPPROTO_UDP);
            v6PAdr->setSockAddr(*SocketOpt::sockaddr_in6_cast(peerSockaddr));
            clientAddr = v6PAdr;
        }
        else {
            SocketOpt::close(clientSockfd);
            return;
        }

        const auto addrStr = clientAddr->toStr();
        const auto ipStr   = addrStr.substr(0, addrStr.find_last_of(":"));
        if (blacklist_.find(ipStr) != blacklist_.end()) {
            SocketOpt::close(clientSockfd);
            TRACE_ERROR << "addr:"_s << addrStr << " in blacklist "_s;
        }
        else {
            if (newConnectionCallback_) {
                newConnectionCallback_(clientSockfd, connId, clientAddr);
            }
            else {
                SocketOpt::close(clientSockfd);
                TRACE_ERROR << "engine:"_s << loopId() << " has no register newConnectionCallback"_s;
            }
        }

        if (n > 0) {  // 首次连接的消息
            for (ssize_t i = 0; i < n;) {
                auto     tmpReadableLen = static_cast<uint32_t>(n - i);
                auto     header         = msg::MessageHeader::decode(buf + i, static_cast<uint32_t>(tmpReadableLen));
                uint32_t msgTotalLen    = header.bodyLen + HEADER_LEN;
                if (tmpReadableLen < msgTotalLen) {
                    break;
                }
                QUEUE_MGR.pubIPCDataToBusiThread(buf + i + sizeof(header), header.bodyLen, header.id, loopId(), connId);
                i += msgTotalLen;
            }
        }
        else if (n == 0) {
            handleClose(shared::getDisconnectedReason(0));
        }
        else {
            auto savedErrno = errno;
            if (savedErrno != EINTR && savedErrno != EAGAIN && savedErrno != EWOULDBLOCK) {
                SYSTEM_ERROR << "Connection::handleRead loop id:"_s << loopId();
                handleError(shared::getDisconnectedReason(savedErrno));
            }
        }
    }
}  // namespace foundation
}  // namespace xncc